package com.ychen.goodscenter.fafka;


import com.ychen.goodscenter.vo.req.SubmitOrderReq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

@Component
@Slf4j
public class MessageProducer {
    @Autowired
    private KafkaTemplate<Long, SubmitOrderReq> kafkaTemplate;

    public void sendOrderMessage(SubmitOrderReq msg) {
        log.info("order-message-topic message sending, orderId: {}", msg.getOrderId());
        kafkaTemplate.send(TopicConstants.ORDER_MESSAGE_TOPIC, msg.getOrderId(), msg);
        log.info("order-message-topic message sent, orderId: {}", msg.getOrderId());
    }

    public void sendOrderMessageSync(SubmitOrderReq msg) {
        CompletableFuture<SendResult<Long, SubmitOrderReq>> future =
                kafkaTemplate.send(TopicConstants.ORDER_MESSAGE_TOPIC, msg.getOrderId(), msg);
        try {
            // 同步等待发送结果
            SendResult<Long, SubmitOrderReq> result = future.get();
            // 处理成功发送的情况
            log.info("order-message-topic message send successfully: " + result.getRecordMetadata().toString());
        } catch (Exception e) {
            // 处理发送失败的情况
            log.info("order-message-topic message send error message: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
}
